package org.ourgrid.peer.controller;



import java.util.Collection;
import java.util.List;

import org.ourgrid.common.interfaces.RemoteWorkerProvider;
import org.ourgrid.common.interfaces.RemoteWorkerProviderClient;
import org.ourgrid.common.interfaces.management.RemoteWorkerManagement;
import org.ourgrid.common.interfaces.management.WorkerManagement;
import org.ourgrid.common.interfaces.to.LocalWorkerState;
import org.ourgrid.common.interfaces.to.RequestSpec;
import org.ourgrid.common.statistics.control.WorkerControl;
import org.ourgrid.peer.PeerConfiguration;
import org.ourgrid.peer.PeerConstants;
import org.ourgrid.peer.controller.allocation.DefaultAllocator;
import org.ourgrid.peer.controller.messages.ConsumerMessages;
import org.ourgrid.peer.controller.messages.RequestMessages;
import org.ourgrid.peer.controller.messages.VOMSMessages;
import org.ourgrid.peer.controller.voms.VomsAuthorisationStrategy;
import org.ourgrid.peer.controller.voms.VomsAuthorisationStrategy.VomsAuthorisationData;
import org.ourgrid.peer.dao.AllocationDAO;
import org.ourgrid.peer.dao.ConsumerDAO;
import org.ourgrid.peer.dao.LocalWorkersDAO;
import org.ourgrid.peer.dao.PeerCertificationDAO;
import org.ourgrid.peer.dao.trust.TrustCommunitiesDAO;
import org.ourgrid.peer.to.AllocableWorker;
import org.ourgrid.peer.to.LocalAllocableWorker;
import org.ourgrid.peer.to.LocalWorker;
import org.ourgrid.peer.to.Priority;
import org.ourgrid.peer.to.RemoteConsumer;
import org.ourgrid.reqtrace.Req;

import br.edu.ufcg.lsd.commune.api.FailureNotification;
import br.edu.ufcg.lsd.commune.api.InvokeOnDeploy;
import br.edu.ufcg.lsd.commune.api.MonitoredBy;
import br.edu.ufcg.lsd.commune.api.RecoveryNotification;
import br.edu.ufcg.lsd.commune.container.servicemanager.ServiceManager;
import br.edu.ufcg.lsd.commune.identification.DeploymentID;
import br.edu.ufcg.lsd.commune.identification.ServiceID;
import br.edu.ufcg.lsd.commune.network.certification.CertificateCRLPair;
import br.edu.ufcg.lsd.commune.network.certification.CertificationUtils;

/**
 * Performs Remote Worker Provider Receiver actions
 */
@Req("REQ011")
public class RemoteWorkerProviderController implements RemoteWorkerProvider{

	private ServiceManager serviceManager;

	@InvokeOnDeploy
	public void init(ServiceManager serviceManager) {
		this.serviceManager = serviceManager;
	}
	
	
	@Req("REQ015")
	public void disposeWorker(ServiceID workerServiceID) {
		
		String consumerPublicKey = serviceManager.getSenderPublicKey();
		
		RemoteConsumer remoteConsumer = serviceManager.getDAO(ConsumerDAO.class).getRemoteConsumer(consumerPublicKey);
		
		if (remoteConsumer == null) {
			serviceManager.getLog().warn(ConsumerMessages.getUnknownConsumerDisposingWorkerMessage(consumerPublicKey));
			return;
		}
		
		DeploymentID remoteConsumerOID = remoteConsumer.getDeploymentID();
		if (workerServiceID == null) {
			serviceManager.getLog().warn(ConsumerMessages.getRemoteConsumerDisposingNullWorkerMessage(remoteConsumerOID));
			return;
		}
		
		String workerPublicKey = workerServiceID.getPublicKey();
		
		LocalWorker localWorker = WorkerControl.getInstance().getRecoveredWorker(
				workerServiceID, serviceManager.getDAO(LocalWorkersDAO.class));
		
		if (localWorker != null && 
				(localWorker.getStatus().equals(LocalWorkerState.OWNER) || localWorker.getStatus().equals(LocalWorkerState.IDLE))) {
			
			serviceManager.getLog().warn(ConsumerMessages.getRemoteConsumerDisposingNotAllocatedWorkerMessage(
					remoteConsumerOID, workerServiceID));
			return;
		}
		
		AllocationDAO allocationDAO = serviceManager.getDAO(AllocationDAO.class);
		AllocableWorker allocable = allocationDAO.getAllocableWorker(workerPublicKey);
		
		if (allocable == null) {
			serviceManager.getLog().warn(ConsumerMessages.getRemoteConsumerDisposingUnknownWorkerMessage(remoteConsumerOID));
			return;
		}
		
		DeploymentID consumerID = allocable.getConsumer().getDeploymentID();
		
		if (!consumerID.getPublicKey().equals(consumerPublicKey)) {
			serviceManager.getLog().warn(ConsumerMessages.getRemoteConsumerDisposingNotAllocatedWorkerMessage(
					remoteConsumerOID, workerServiceID));
			return;
		}
		
		boolean workerIsUp = serviceManager.getStubDeploymentID(allocable.getWorkerManagement()) != null;
		
		if (workerIsUp) {
			WorkerManagement workerManagement= (WorkerManagement) allocable.getWorkerManagement();
			
			serviceManager.getLog().debug(ConsumerMessages.getRemoteClientDisposingWorkerMessage(remoteConsumerOID, workerServiceID));
			workerManagement.stopWorking();
			
			Object consumerStub = allocable.getConsumer().getConsumerStub();
			allocable.deallocate();
			
			WorkerControl.getInstance().statusChanged(
					allocable.getWorkerSpec().getServiceID(), 
					LocalWorkerState.IDLE, serviceManager);
			
			if (remoteConsumer.getAllocableWorkers().isEmpty()) {
				serviceManager.getDAO(ConsumerDAO.class).removeRemoteConsumer(consumerPublicKey);
				serviceManager.release(consumerStub);
			}

		} else {
			WorkerNotificationController.workerFailure(serviceManager, workerServiceID, localWorker);
		}
		
	}


	@RecoveryNotification
	public void remoteWorkerManagementIsUp(RemoteWorkerManagement worker) {}

	@FailureNotification
	public void remoteWorkerManagementIsDown(RemoteWorkerManagement worker) {}

	/**
	 * Validates and executes a remote request.
	 * 
	 * @param workerProviderClient 
	 * @param requestSpec
	 * @param senderPublicKey
	 */
	@Req("REQ011")
	public void requestWorkers(
			@MonitoredBy(PeerConstants.REMOTE_WORKER_PROVIDER_CLIENT_MONITOR)
			RemoteWorkerProviderClient workerProviderClient, RequestSpec requestSpec) {
		
		String senderPublicKey = serviceManager.getSenderPublicKey();
		
		if(senderPublicKey == null) {
			serviceManager.getLog().warn(RequestMessages.getRequestWithNoPublicKeyMessage(requestSpec));
			serviceManager.release(workerProviderClient);
			return;
		}
		
		if(workerProviderClient == null) {
			serviceManager.getLog().warn(RequestMessages.getRequestWithNoClientMessage(requestSpec));
//			serviceManager.release(workerProviderClient);
			return;
		}
		
		DeploymentID workerProviderClientID = serviceManager.getStubDeploymentID(workerProviderClient);
		
		if(requestSpec == null) {
			serviceManager.getLog().warn(RequestMessages.getNullRequestSpecMessage(workerProviderClientID));
			serviceManager.release(workerProviderClient);
			return;
		}
		
		if(requestSpec.getRequiredWorkers() < 1) {
			serviceManager.getLog().warn(RequestMessages.getNonPositiveWorkerRequestMessage(workerProviderClientID,
					requestSpec.getRequestId()));
			serviceManager.release(workerProviderClient);
			return;
		}
		
		
		//authentication
		Collection<CertificateCRLPair> requestingPeersCAsData = serviceManager.getDAO(PeerCertificationDAO.class).getRequestingPeersCAsData();
		
		if(requestingPeersCAsData != null && !requestingPeersCAsData.isEmpty()) {
			
			if(!CertificationUtils.isCertificateValid(requestSpec.getRequesterCertPath())) {
				serviceManager.getLog().warn(RequestMessages.getInvalidCertPathMessage(workerProviderClientID,
						requestSpec.getRequestId()));
				serviceManager.release(workerProviderClient);
				return;
			}
			
			if(!CertificationUtils.isCertPathIssuedByCA(requestSpec.getRequesterCertPath(), requestingPeersCAsData)) {
				serviceManager.getLog().warn(RequestMessages.getNonIssuedCertPathMessage(workerProviderClientID,
						requestSpec.getRequestId()));
				serviceManager.release(workerProviderClient);
				return;
			}
			
		}
		
		VomsAuthorisationData authorisationData = null;
		
		//authorization
		if(serviceManager.getContainerContext().isEnabled(PeerConfiguration.PROP_USE_VOMS)) {
			VomsAuthorisationStrategy authStrategy = new VomsAuthorisationStrategy(serviceManager);
			
			try {
				authorisationData = authStrategy.authorise(CertificationUtils.getCertSubjectDN(
						requestSpec.getRequesterCertPath()));
			} catch (Exception e) {
				serviceManager.getLog().warn(VOMSMessages.getErrorOnConnectingToVOMSMessage(workerProviderClientID,
						requestSpec.getRequestId(), e.getMessage()));
				serviceManager.release(workerProviderClient);
				return;
			}
			
			if (!authorisationData.isAuthorised()) {
				serviceManager.getLog().warn(VOMSMessages.getNonAuthorisedConsumerMessage(workerProviderClientID,
						requestSpec.getRequestId()));
				serviceManager.release(workerProviderClient);
				return;
			}
		}
		
		serviceManager.getLog().info(RequestMessages.getRequestWorkersMesasge(workerProviderClientID, requestSpec));

		doRequest(requestSpec, workerProviderClient, authorisationData);
	}

	/**
	 * Executes a request. This method does not validate the request.
	 * @param requestSpec 
	 * @param workerProviderClient
	 * @param authorisationData 
	 */
	@Req("REQ011")
	private void doRequest(RequestSpec requestSpec, RemoteWorkerProviderClient workerProviderClient, 
			VomsAuthorisationData authorisationData) {
		
		DeploymentID workerProviderClientID = serviceManager.getStubDeploymentID(workerProviderClient);
		RemoteConsumer consumer = serviceManager.getDAO(ConsumerDAO.class).getRemoteConsumer(
				workerProviderClientID.getPublicKey());
		
		if (consumer == null) {
			consumer = new RemoteConsumer();
			consumer.setConsumerStub(workerProviderClientID, workerProviderClient);
			consumer.setConsumerDN(CertificationUtils.getCertSubjectDN(serviceManager.getSenderCertPath()));
			
			Priority priority = createPriority(workerProviderClientID.getServiceID());
			consumer.setPriority(priority);
			
			serviceManager.getDAO(ConsumerDAO.class).addRemoteConsumer(workerProviderClientID.getPublicKey(), consumer);
		}
		
		if (!workerProviderClientID.equals(consumer.getDeploymentID())) {
			consumer.setConsumerStub(workerProviderClientID, workerProviderClient);
		}
		
		List<AllocableWorker> allocableWbag = serviceManager.getDAO(AllocationDAO.class).getLocalAllocableWorkers();
		
		//try to serve the request
		List<AllocableWorker> allocation = DefaultAllocator.getInstance().
												getAllocableWorkersForRemoteRequest(
														consumer, requestSpec, allocableWbag, serviceManager);
		
				
		if(! allocation.isEmpty()){

			serviceManager.registerInterest(PeerConstants.REMOTE_WORKER_PROVIDER_CLIENT_MONITOR, 
					workerProviderClientID.getServiceID().toString(), RemoteWorkerProviderClient.class);
			
			for (AllocableWorker allocableWorker : allocation) {
				dispatchAllocation(requestSpec.getRequestId(), workerProviderClient, 
						consumer, allocableWorker, authorisationData);
			}
			
		} else if (consumer.getAllocableWorkers().isEmpty()){
			serviceManager.release(workerProviderClient);
		}
		
	}

	/**
	 * @param requestID 
	 * @param workerProviderClient
	 * @param consumer
	 * @param allocableWorker
	 * @param authorisationData 
	 */
	private void dispatchAllocation( long requestID, RemoteWorkerProviderClient workerProviderClient,
			RemoteConsumer consumer, AllocableWorker allocableWorker, VomsAuthorisationData authorisationData) {
		
		if(allocableWorker.getConsumer() != null) {//preemption
			DeploymentID workerID = serviceManager.getStubDeploymentID(allocableWorker.getWorkerManagement());
			
			serviceManager.getLog().info(RequestMessages.getRequestPreemptionMessage(requestID, 
					workerID, 
					allocableWorker.getConsumer().getDeploymentID()));
		}
		
		DeploymentID workerProviderClientID = serviceManager.getStubDeploymentID(workerProviderClient);
		
		allocableWorker.clear();
		allocableWorker.setStatus(LocalWorkerState.DONATED);
		
		WorkerControl.getInstance().statusChanged(
				allocableWorker.getWorkerSpec().getServiceID(), 
				LocalWorkerState.DONATED, 
				consumer.getDeploymentID().getContainerID().getUserAtServer(), serviceManager);
		
		allocableWorker.setConsumer(consumer);
		consumer.addWorker(allocableWorker);
		LocalAllocableWorker localAllocableWorker = (LocalAllocableWorker)allocableWorker;
		
		if (authorisationData == null) {
			localAllocableWorker.workForPeer(workerProviderClientID.getPublicKey());
		} else {
			localAllocableWorker.workForPeer(workerProviderClientID.getPublicKey(), 
					authorisationData.getUsersDN(), 
					serviceManager.getDAO(PeerCertificationDAO.class).getRequestingPeersCAsCertificates());
		}
		
	}

	/**
	 * @return
	 */
	private Priority createPriority(ServiceID entityID) {
		return serviceManager.getDAO(TrustCommunitiesDAO.class).getPriority(
				entityID.getPublicKey());
	}
}